# Databricks notebook source
# MAGIC %md
# MAGIC ## TMR Populate Case Dimensions
# MAGIC <p><b>Description: </b>This notebook aims to populate all Case dimensions needed for the TMR reports A.K.A Developer Services.</p>
# MAGIC <b>Parent Process: </b>PL_FactPrism
# MAGIC <table class="table table-striped table-hover">
# MAGIC  <thead>
# MAGIC   <tr>
# MAGIC    <th>Contributor</th>
# MAGIC    <th>Date</th>
# MAGIC    <th>Version</th>
# MAGIC    <th>Comment</th>
# MAGIC    <th>WorkItem No</th>
# MAGIC   </tr>
# MAGIC  </thead>
# MAGIC  <tbody>
# MAGIC   <tr>
# MAGIC    <td>Iwan Dyke</td>
# MAGIC    <td>18/04/2024</td>
# MAGIC    <td>1.0</td>
# MAGIC    <td>Create initial release</td>
# MAGIC    <td></td>
# MAGIC   </tr>
# MAGIC  </tbody>
# MAGIC </table>

# COMMAND ----------

# MAGIC %md
# MAGIC ## Import Python Libraries

# COMMAND ----------

from pyspark.sql.types import *
from pyspark.sql.functions import *
from pyspark.sql.window import Window
import pandas as pd  
import numpy as np 
import os  # Joining filenames.
import datetime as dt  # Adding today's date to exports.
from datetime import datetime
import glob

# COMMAND ----------

# MAGIC %md
# MAGIC ## Populating Type 0 Task Dimensions

# COMMAND ----------

# Checking type 0 dimension dim_task_status

dim_case_status_sql_check = """
WITH CTE_dim_case_status_values AS (
SELECT DISTINCT 
  s.`Case Status`
FROM 
	databricks_poc.staging.tmr_raw s
	LEFT OUTER JOIN databricks_poc.tmr.dim_case_status d
	ON s.`Case Status` = d.Description  
WHERE d.Description IS NULL
)

SELECT ALL * FROM CTE_dim_case_status_values
"""

dim_case_status_data_check = spark.sql(dim_case_status_sql_check)

if dim_case_status_data_check.count() > 0:
  raise ValueError("New data found in dim_case_status")

# COMMAND ----------

# Checking type 0 dimension dim_lpa

dim_channel_sql_check = """
WITH CTE_dim_channel_values AS (
SELECT DISTINCT 
  s.`Channel`
FROM 
	databricks_poc.staging.tmr_raw s
	LEFT OUTER JOIN databricks_poc.tmr.dim_channel d
	ON s.`channel` = d.`Description`
WHERE d.Description IS NULL
)

SELECT ALL * FROM CTE_dim_channel_values
WHERE channel IS NOT NULL
"""

dim_channel_data_check = spark.sql(dim_channel_sql_check)

if dim_channel_data_check.count() > 0:
  raise ValueError("New data found in dim_channel")

# COMMAND ----------

# MAGIC %md
# MAGIC ## Populating Type 2 Task Dimensions

# COMMAND ----------

# MAGIC %sql
# MAGIC -- Delete records from Dim_CaseEngineer
# MAGIC DELETE FROM databricks_poc.tmr.dim_case_engineer
# MAGIC WHERE Code > 0;
# MAGIC
# MAGIC -- Capture all technical assessments
# MAGIC CREATE OR REPLACE TEMPORARY VIEW ProjectEngineer AS
# MAGIC (
# MAGIC SELECT 
# MAGIC     `BPM Reference Number` AS BPMReference,
# MAGIC     `Task Instance ID` AS TaskInstanceID,
# MAGIC     CAST(`Date Started` AS DATE) AS CaseStartDate,
# MAGIC     0 AS OriginalEngineer,
# MAGIC     CAST(`Date Task Created` AS DATE) AS OpenDate,
# MAGIC     CAST(NULL AS DATE) AS CloseDate,
# MAGIC     CASE 
# MAGIC         WHEN LENGTH(`Task Owner`) = 0 THEN 'Unassigned'
# MAGIC         ELSE `Task Owner`
# MAGIC     END AS TaskOwner,
# MAGIC     ROW_NUMBER() OVER (PARTITION BY `BPM Reference Number` ORDER BY `Date Task Created`, `Task Instance ID`) AS Ordering
# MAGIC FROM databricks_poc.staging.tmr_raw s
# MAGIC WHERE --`Task Subject(0-60)` LIKE 'Technical assessment%'
# MAGIC   `Task Owner` NOT LIKE 'Test%'
# MAGIC    AND LENGTH(`Task Owner`) > 0
# MAGIC );
# MAGIC
# MAGIC -- Derive CloseDate for original records
# MAGIC CREATE OR REPLACE TEMPORARY VIEW ProjectEngineerOriginalRecords AS
# MAGIC (
# MAGIC SELECT 
# MAGIC     o1.BPMReference,
# MAGIC     o1.TaskInstanceID,
# MAGIC     o1.CaseStartDate,
# MAGIC     CASE
# MAGIC         WHEN o1.Ordering = 1 THEN 1
# MAGIC         ELSE o1.OriginalEngineer
# MAGIC     END AS OriginalEngineer,
# MAGIC     o1.OpenDate,
# MAGIC     o2.OpenDate AS CloseDate,
# MAGIC     o1.TaskOwner,
# MAGIC     o1.Ordering
# MAGIC FROM 
# MAGIC     ProjectEngineer o1
# MAGIC     JOIN ProjectEngineer o2
# MAGIC     ON o1.BPMReference = o2.BPMReference
# MAGIC     AND o1.Ordering = o2.Ordering - 1
# MAGIC );
# MAGIC
# MAGIC -- Create temporary view for updated CloseDate (in ProjectEngineer) from original records
# MAGIC CREATE OR REPLACE TEMPORARY VIEW ProjectEngineerUpdated AS
# MAGIC (
# MAGIC SELECT 
# MAGIC     o.`BPMReference`,
# MAGIC     o.`TaskInstanceID`,
# MAGIC     o.`CaseStartDate`,
# MAGIC     CASE
# MAGIC         WHEN c.`OpenDate` = c.`CloseDate` THEN c.`CloseDate`
# MAGIC         ELSE DATE_SUB(c.`CloseDate`, 1)
# MAGIC     END AS `CloseDate`,
# MAGIC     CASE
# MAGIC         WHEN o.`Ordering` = 1 THEN 1
# MAGIC         ELSE o.`OriginalEngineer`
# MAGIC     END AS `OriginalEngineer`,
# MAGIC     o.`OpenDate`,
# MAGIC     o.`TaskOwner`,
# MAGIC     o.`Ordering`
# MAGIC FROM 
# MAGIC     ProjectEngineer o
# MAGIC     JOIN ProjectEngineerOriginalRecords c
# MAGIC     ON o.`TaskInstanceID` = c.`TaskInstanceID`
# MAGIC );
# MAGIC
# MAGIC -- Insert into Dim_CaseEngineer from ProjectEngineerUpdated
# MAGIC INSERT INTO databricks_poc.tmr.dim_case_engineer
# MAGIC (
# MAGIC     BPMReference
# MAGIC     ,TaskOwner
# MAGIC     ,CaseStartDate
# MAGIC     ,OriginalEngineer
# MAGIC     ,OpenDate
# MAGIC     ,ClosedDate
# MAGIC     ,CurrentFlag
# MAGIC )
# MAGIC SELECT 
# MAGIC     `BPMReference`,
# MAGIC     `TaskOwner`,
# MAGIC     `CaseStartDate`,
# MAGIC     `OriginalEngineer`,
# MAGIC     `OpenDate`,
# MAGIC     `CloseDate`,
# MAGIC     CASE
# MAGIC         WHEN `CloseDate` IS NULL THEN 1
# MAGIC         ELSE 0
# MAGIC     END AS `CurrentFlag`
# MAGIC FROM 
# MAGIC     ProjectEngineerUpdated;
# MAGIC
